package com.boarsoft.boar.message.service.impl;

import java.util.Map;
import java.util.Map.Entry;
import java.util.Vector;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import javax.annotation.PostConstruct;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import com.boarsoft.boar.message.adapter.MessageBrokerAdapter;
import com.boarsoft.boar.message.service.QueueMonitor;
import com.boarsoft.common.Util;
import com.boarsoft.message.bean.MessageQueueStat;

public abstract class AbstractQueueMonitorImpl implements QueueMonitor, Runnable {
	private static final Logger log = LoggerFactory.getLogger(AbstractQueueMonitorImpl.class);

	@Autowired
	protected MessageBrokerAdapter brokerAdapter;

	/** */
	protected ScheduledExecutorService scheduler;
	/** */
	protected Vector<String> moningSet = new Vector<String>();
	/** */
	protected long period = 3L;

	/** */
	protected String type;

	/** queueId queue stat map */
	protected Map<String, MessageQueueStat> queueStatMap = //
			new ConcurrentHashMap<String, MessageQueueStat>();

	@PostConstruct
	public void init() {
		scheduler.scheduleWithFixedDelay(this, period, period, TimeUnit.SECONDS);
	}

	@Override
	public void run() {
		if (moningSet == null || moningSet.isEmpty()) {
			return;
		}
		for (String t : moningSet) {
			try {
				this.stat(t);
			} catch (Exception e) {
				log.error("Error on get stat of queue {}", t, e);
			}
		}
	}

	protected abstract Object gather(String queueId);

	@SuppressWarnings("unchecked")
	protected void stat(String queueId) {
		// 从多个Broker获取这个队列的信息，返回的是一个Map
		Object ro = this.gather(queueId);
		Map<String, MessageQueueStat> rm = (Map<String, MessageQueueStat>) ro;
		int in = 0, out = 0, size = 0;
		for (Entry<String, MessageQueueStat> en : rm.entrySet()) {
			String k = en.getKey();
			MessageQueueStat o = en.getValue();
			if (o == null) {
				log.warn("Can not get {} queue {} stat from {}", type, queueId, k);
				continue;
			}
			size += o.getSize();
			in += o.getIn();
			out += o.getOut();
			log.info("Got {} queue {} stat {}/{}/{} from {}",//
					type, queueId, o.getSize(), o.getIn(), o.getOut(), k);
		}
		//
		if (Util.strIsEmpty(queueId)) {
			queueId = "overall";
		}
		MessageQueueStat so = queueStatMap.get(queueId);
		long l = System.currentTimeMillis();
		if (so == null) {
			so = new MessageQueueStat();
		} else {
			// 计算时间间隔，以计算出入队速度
			int i = new Long(l - so.getTime()).intValue();
			i = Math.max(1, i / 1000);
			so.setIn(in / i);
			so.setOut(out / i);
			so.setTime(l);
		}
		so.setSize(size);
		queueStatMap.put(queueId, so);
	}

	@Override
	public MessageQueueStat getStat(String broker, String queueId) {
		if (!moningSet.contains(queueId)) {
			synchronized (moningSet) {
				if (!moningSet.contains(queueId)) {
					moningSet.add(queueId);
				}
			}
		}
		if (queueStatMap.containsKey(queueId)) {
			return queueStatMap.get(queueId);
		}
		synchronized (queueStatMap) {
			if (queueStatMap.containsKey(queueId)) {
				return queueStatMap.get(queueId);
			}
			MessageQueueStat r = new MessageQueueStat();
			queueStatMap.put(queueId, r);
			return r;
		}
	}

	public ScheduledExecutorService getScheduler() {
		return scheduler;
	}

	public void setScheduler(ScheduledExecutorService scheduler) {
		this.scheduler = scheduler;
	}

	@Override
	public Vector<String> getMoningSet() {
		return moningSet;
	}

	public void setMoningSet(Vector<String> moningSet) {
		this.moningSet = moningSet;
	}

	public String getType() {
		return type;
	}

	public void setType(String type) {
		this.type = type;
	}

	public long getPeriod() {
		return period;
	}

	public void setPeriod(long period) {
		this.period = period;
	}
}
